package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/louloulin/dataflare/pkg/core"
	_ "github.com/louloulin/dataflare/pkg/connector/console"
	_ "github.com/louloulin/dataflare/pkg/connector/memory"
	_ "github.com/louloulin/dataflare/pkg/processor/transform"
)

func main() {
	fmt.Println("DataFlare New Workflow Example")
	fmt.Println("-----------------------------")

	// 创建系统
	system, err := core.NewSystem()
	if err != nil {
		fmt.Printf("Failed to create system: %v\n", err)
		os.Exit(1)
	}

	// 启动系统
	ctx := context.Background()
	err = system.Start(ctx)
	if err != nil {
		fmt.Printf("Failed to start system: %v\n", err)
		os.Exit(1)
	}
	defer system.Stop(ctx)

	// 创建工作流引擎
	engine := core.NewWorkflowEngine(system)

	// 创建工作流定义
	definition := &core.WorkflowDefinition{
		ID:          "new-workflow",
		Name:        "New Workflow",
		Description: "A new workflow example",
		Version:     "1.0",
		Sources: map[string]core.SourceDefinition{
			"memory-source": {
				Type: "memory",
				Config: map[string]interface{}{
					"data": []map[string]interface{}{
						{"id": "1", "name": "Alice", "age": 30},
						{"id": "2", "name": "Bob", "age": 25},
						{"id": "3", "name": "Charlie", "age": 35},
					},
				},
			},
		},
		Processors: map[string]core.ProcessorDefinition{
			"transform-processor": {
				Type: "transform",
				Config: map[string]interface{}{
					"function": `
						function process(data) {
							const parsed = JSON.parse(data);
							if (parsed.age < 30) {
								parsed.category = "young";
							} else {
								parsed.category = "adult";
							}
							return JSON.stringify(parsed);
						}
					`,
				},
			},
		},
		Sinks: map[string]core.SinkDefinition{
			"console-sink": {
				Type: "console",
				Config: map[string]interface{}{
					"format": "json",
				},
			},
		},
		Edges: []core.EdgeDefinition{
			{
				From: "memory-source",
				To:   "transform-processor",
			},
			{
				From: "transform-processor",
				To:   "console-sink",
			},
		},
	}

	// 部署工作流
	fmt.Println("Deploying workflow...")
	workflowID, err := engine.DeployWorkflow(ctx, definition)
	if err != nil {
		fmt.Printf("Failed to deploy workflow: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Workflow deployed with ID: %s\n", workflowID)

	// 启动工作流
	fmt.Println("Starting workflow...")
	runID, err := engine.StartWorkflow(ctx, workflowID, nil)
	if err != nil {
		fmt.Printf("Failed to start workflow: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Workflow started with run ID: %s\n", runID)

	// 等待工作流完成或用户中断
	fmt.Println("Workflow is running. Press Ctrl+C to stop...")
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			// 获取工作流状态
			status, err := engine.GetWorkflowStatus(ctx, runID)
			if err != nil {
				fmt.Printf("Failed to get workflow status: %v\n", err)
				continue
			}

			fmt.Printf("Workflow status: %s (Progress: %d%%)\n", status.Status, status.Progress)

			if status.Status == "completed" || status.Status == "failed" {
				fmt.Printf("Workflow %s\n", status.Status)
				return
			}

		case <-sigCh:
			fmt.Println("Stopping workflow...")
			err := engine.StopWorkflow(ctx, runID)
			if err != nil {
				fmt.Printf("Failed to stop workflow: %v\n", err)
				os.Exit(1)
			}

			fmt.Println("Workflow stopped successfully")
			return
		}
	}
}
